package com.pw.study.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.pw.study.flink.entities.UserBean
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}

class MyJdbcSink() extends RichSinkFunction[UserBean] {
  var connection: Connection = _
  var ps: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    println("open...............")
    Class.forName("com.mysql.jdbc.Driver")
    connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/study", "root", "123456")
    val sql: String = "insert into tb_vc(id,ts,vc) values(?,?,?) on duplicate key update id=?"
    ps = connection.prepareStatement(sql)

  }

  override def invoke(value: UserBean, context: SinkFunction.Context[_]): Unit = {
    println("value: " + value)
    ps.setString(1,value.id)
    ps.setString(2,value.ts+"")
    ps.setString(3,value.vc+"")
    ps.setString(4,value.id)
    ps.execute()

  }

  override def close(): Unit = {
    println("close................")
    ps.close()
    connection.close()
  }
}
